package com.zyx.restaurant.server.service.impl;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.*;
import com.zyx.restaurant.server.mapper.ProductMapper;
import com.zyx.restaurant.server.mapper.RestaurantMapper;
import com.zyx.restaurant.server.model.dto.OrderMessageDTO;
import com.zyx.restaurant.server.common.enums.ProductStatus;
import com.zyx.restaurant.server.common.enums.RestaurantStatus;
import com.zyx.restaurant.server.model.po.ProductPO;
import com.zyx.restaurant.server.model.po.RestaurantPO;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author Yaxi.Zhang
 * @since 2022/10/22 07:58
 */
@Slf4j
@Service
public class OrderMessageService {

    @Autowired
    private ObjectMapper objectMapper;

    @Autowired
    private ProductMapper productMapper;

    @Autowired
    private RestaurantMapper restaurantMapper;

    @Autowired
    private Channel channel;

    @Async
    public void handleMessage() throws IOException, TimeoutException, InterruptedException {
        log.info("start linstening message");
        // ConnectionFactory connectionFactory = new ConnectionFactory();
        // connectionFactory.setHost("81.68.117.201");
        // try (Connection connection = connectionFactory.newConnection();
        //      Channel channel = connection.createChannel()) {
        //
        //     this.channel = channel;

        // 声明接收死信队列的交换机及队列
        // 注意: 这里声明的并不是死信队列, 而是接收死信队列的队列,
        //      死信队列是添加了 x-dead-letter-exchange 参数的队列
        //      死信队列会将超过了过期时间及队列容量的消息丢弃到该队列中, 供异常处理类处理
        channel.exchangeDeclare("exchange.dlx", BuiltinExchangeType.TOPIC, true, false, null);
        channel.queueDeclare("queue.dlx", true, false, false, null);
        channel.queueBind("queue.dlx", "exchange.dlx", "#");

        channel.exchangeDeclare("exchange.order.restaurant", BuiltinExchangeType.DIRECT, true, false, null);

        // 设置参数
        // 记得先去RabbitMQ删除原来的队列, 否则声明会报错
        Map<String, Object> args = new HashMap<>(1 << 4);
        // 队列的消息过期时间
        args.put("x-message-ttl", 15000);
        // 与下面的参数进行区分: 下面参数指队列的删除时间, 表示一个队列一段时间没有收到消息后, 直接将该队列删除, 不建议开启!!!
        // args.put("x-expire", 15000);
        // 设置队列的最大长度
        args.put("x-max-length", 5);
        // 设置死信队列
        args.put("x-dead-letter-exchange", "exchange.dlx");

        channel.queueDeclare("queue.restaurant", true, false, false, args);

        // channel.queueDeclare("queue.restaurant", true, false, false, null);
        channel.queueBind("queue.restaurant", "exchange.order.restaurant", "key.restaurant");

        // 指定消费端限流
        channel.basicQos(2);

        // channel.basicConsume("queue.restaurant", true, deliverCallback, consumerTag -> {
        // });
        // 关闭自动ack
        channel.basicConsume("queue.restaurant", false, deliverCallback, consumerTag -> {
        });
        while (true) {
            Thread.sleep(100000);
        }
    }
    // }

    DeliverCallback deliverCallback = (consumerTag, message) -> {
        String messageBody = new String(message.getBody());
        log.info("deliverCallback:messageBody:{}", messageBody);
        // ConnectionFactory connectionFactory = new ConnectionFactory();
        // connectionFactory.setHost("81.68.117.201");
        try {
            OrderMessageDTO orderMessageDTO = objectMapper.readValue(messageBody, OrderMessageDTO.class);
            ProductPO productPO = productMapper.selectById(orderMessageDTO.getProductId());
            log.info("onMessage:productPO:{}", productPO);
            RestaurantPO restaurantPO = restaurantMapper.selectById(productPO.getRestaurantId());
            log.info("onMessage:restaurantPO:{}", restaurantPO);
            if (ProductStatus.AVAILABLE == productPO.getStatus() && RestaurantStatus.OPEN == restaurantPO.getStatus()) {
                orderMessageDTO.setConfirmed(true);
                orderMessageDTO.setPrice(productPO.getPrice());
            } else {
                orderMessageDTO.setConfirmed(false);
            }
            log.info("sendMessage:restaurantOrderMessageDTO:{}", orderMessageDTO);
            // AutoClosable


            // try (Connection connection = connectionFactory.newConnection();
            //      Channel channel = connection.createChannel()) {


            // channel.addReturnListener(new ReturnListener() {
            //     @Override
            //     public void handleReturn(int replyCode, String replyText,
            //                              String exchange, String routingKey, AMQP.BasicProperties properties, byte[] body) throws IOException {
            //         log.info("Message Return: replyCode:{}, replyText:{}, exchange:{}, routingKey:{}, properties:{}, body:{}",
            //                 replyCode, replyText, exchange, routingKey, properties, new String(body));
            //         // 除了打印log，可加家别的业务操作
            //     }
            // });

            // 封装成对象返回
            channel.addReturnListener(new ReturnCallback() {
                @Override
                public void handle(Return returnMessage) {
                    log.info("Message Return: returnMessage:{}", returnMessage);
                    // 除了打印log，可加家别的业务操作
                }
            });

            // 模拟消费处理速度慢, 演示限流机制作用
            TimeUnit.SECONDS.sleep(1L);

            // 如果关闭了自动ack, 则必须要手动ack, 否则无法正常消费!!!
            // 消费端单条手动ACK
            channel.basicAck(message.getEnvelope().getDeliveryTag(), false);
            // 消费端多条手动ACK
            // if (message.getEnvelope().getDeliveryTag() % 5 == 0) {
            //     channel.basicAck(message.getEnvelope().getDeliveryTag(),true);
            // }

            // 手动拒收并重回队列, 可能会造成消费不停地重回队列
            // 不建议开启重回队列, 因为当签收失败时, 重新发送大概率还是签收失败, 开启重回队列可能会造成死循环
            // channel.basicNack(message.getEnvelope().getDeliveryTag(), false, true);

            String messageToSend = objectMapper.writeValueAsString(orderMessageDTO);
            // channel.basicPublish("exchange.order.restaurant", "key.order", null, messageToSend.getBytes());
            // 指定重载方法第三个参数 mandatory 为 true，开启托管
            // RabbitMQ会调用 ReturnListener 中的方法处理无法路由的消息
            channel.basicPublish(
                    "exchange.order.restaurant", "key.order", true, null, messageToSend.getBytes());

            // 模拟 ReturnListener , 防止 AutoClosable
            // TimeUnit.SECONDS.sleep(1L);
            // }
        } catch (JsonProcessingException | /*TimeoutException |*/ InterruptedException e) {
            e.printStackTrace();
        }
    };
}